use std::collections::HashMap;
use std::net::SocketAddr;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use bytes::Bytes;
use xor_name::XorName;
use qp2p::{Connection, ConnectionIncoming};
use tokio::sync::mpsc::Sender;
use tokio::sync::{Mutex, RwLock};
use tracing::{debug, error, info, trace, warn};
use crate::Attribute;

use crate::error::*;
use crate::network::{Event, Request};

const REMOTE_TIMEOUT: u128 = 2000;

enum Response {
    Open {
        id: u64,
        fh: u64,
        code: u32,
    },
    Read {
        id: u64,
        fh: u64,
        code: u32,
        data: Bytes,
    },
    Write {
        id: u64,
        fh: u64,
        code: u32,
        size: u32,
    },
    Release {
        id: u64,
        fh: u64,
        code: u32,
    },
}

#[derive(Clone)]
pub struct NodeRemote {
    generator: Arc<Mutex<snowflake::SnowflakeIdGenerator>>,
    waiting: Arc<RwLock<HashMap<u64, (SystemTime, Sender<Response>)>>>,
    running: Arc<AtomicBool>,
    address: SocketAddr,
    sender: Sender<Event>,
    peer: XorName,
}

impl NodeRemote {
    pub fn run(peer: XorName, connection: Connection, mut incoming: ConnectionIncoming, sender: Sender<Request>, generator: Arc<Mutex<snowflake::SnowflakeIdGenerator>>) -> Self {
        let address = connection.remote_address();
        let running = Arc::new(AtomicBool::new(true));
        let (remote, mut receiver) = tokio::sync::mpsc::channel::<Event>(256);
        let waiting = Arc::new(RwLock::new(HashMap::<u64, (SystemTime, Sender<Response>)>::new()));

        let node = remote.clone();
        let peer_remote = peer.clone();

        let waiters = waiting.clone();
        let remote_address = address.clone();
        let receiving = running.clone();
        tokio::spawn(async move {
            while receiving.load(Ordering::SeqCst) {
                match incoming.next().await {
                    Ok(income) => {
                        if let Some(bytes) = income {
                            let event: Event = bincode::deserialize(&bytes).unwrap();
                            match event {
                                Event::Ping {id} => {
                                    if let Err(e) = node.send(Event::Pong {id}).await {
                                        error!("send pong error >>>>> {:?}", e);
                                    } else {
                                        trace!("pong event sent. {}", id);
                                    }
                                },
                                Event::Pong {id} => debug!("receive pong: {}", id),
                                Event::Metadata(nodes) => {
                                    trace!("receive metadata from {}: {}", peer_remote, nodes.len());
                                    let message = Request::Metadata {data: nodes, peer: peer_remote};
                                    if let Err(e) = sender.send(message).await {
                                        error!("send {} metadata error: {}", peer, e);
                                    }
                                }
                                Event::Contacts(addr) => {
                                    trace!("receive contacts from {}: {}", peer_remote, &addr);
                                    let message = Request::Contacts {data: addr, peer: peer_remote};
                                    if let Err(e) = sender.send(message).await {
                                        error!("send {} contacts error: {}", peer, e);
                                    }
                                }
                                Event::Create((path, attribute)) => {
                                    trace!("receive create event from {}: {}", peer_remote, path.display());
                                    let message = Request::Create {data: (path, attribute), peer: peer_remote};
                                    if let Err(e) = sender.send(message).await {
                                        error!("send {} create event error: {}", peer, e);
                                    }
                                }
                                Event::Rename {source, target} => {
                                    trace!("receive create event from {}: {} => {}", peer_remote, source.display(), target.display());
                                    let message = Request::Rename {source, target, peer: peer_remote};
                                    if let Err(e) = sender.send(message).await {
                                        error!("send {} rename event error: {}", peer, e);
                                    }
                                }
                                Event::Open {id, path} => {
                                    debug!("receive open event {}, {}", id, path.display());
                                    let message = Request::Open {id, path, peer};
                                    if let Err(e) = sender.send(message).await {
                                        error!("send open event local error: {}", e);
                                    }
                                }
                                Event::Read {id, fh, size, offset} => {
                                    trace!("receive read event {}, {}", id, fh);
                                    let message = Request::Read {id, fh, size, offset, peer};
                                    if let Err(e) = sender.send(message).await {
                                        error!("send read event local error: {}", e);
                                    }
                                }
                                Event::Write {id, fh, data, offset} => {
                                    trace!("receive write event {}, {}, {}", id, fh, offset);
                                    let message = Request::Write {id, fh, offset, data, peer};
                                    if let Err(e) = sender.send(message).await {
                                        error!("send write event local error: {}", e);
                                    }
                                }
                                Event::Remove(path) => {
                                    trace!("receive remove event, path {}", path.display());
                                    let message = Request::Remove {path, peer};
                                    if let Err(e) = sender.send(message).await {
                                        error!("send remove event local error: {}", e);
                                    }
                                }
                                Event::Rmdir(path) => {
                                    trace!("receive rmdir event, path {}", path.display());
                                    let message = Request::Rmdir {path, peer};
                                    if let Err(e) = sender.send(message).await {
                                        error!("send rmdir event local error: {}", e);
                                    }
                                }
                                Event::Release {id, fh} => {
                                    trace!("receive release event {}, {}", id, fh);
                                    let message = Request::Release {id, fh, peer};
                                    if let Err(e) = sender.send(message).await {
                                        error!("send release event local error: {}", e);
                                    }
                                }
                                Event::Opened {id, fh, code} => {
                                    debug!("receive opened event {}, {}, {}", id, fh, code);
                                    let message = Response::Open {id, fh, code};
                                    response(id, message, waiters.clone()).await;
                                }
                                Event::ReadResponse {id, fh, code, data} => {
                                    let message = Response::Read{id, fh, code, data};
                                    response(id, message, waiters.clone()).await;
                                }
                                Event::WriteResponse {id, fh, code, size} => {
                                    let message = Response::Write{id, fh, code, size};
                                    response(id, message, waiters.clone()).await;
                                }
                                Event::Released {id, fh, code} => {
                                    let message = Response::Release {id, fh, code};
                                    response(id, message, waiters.clone()).await;
                                }
                                Event::Attribute { path, attr: attribute } => {
                                    let message = Request::Attribute {path, attr: attribute, peer};
                                    if let Err(e) = sender.send(message).await {
                                        error!("send attribute async event local error: {}", e);
                                    }
                                }
                            }
                        }
                    }
                    Err(e) => {
                        error!("receive data from remote {} error: {:?}", remote_address, e);
                        receiving.store(false, Ordering::SeqCst);
                    }
                }
            }
            warn!("receive endpoint exit ------------------------------");
        });

        let now = SystemTime::now();
        let dur = now.duration_since(UNIX_EPOCH).unwrap();
        let now = Arc::new(AtomicU64::new(dur.as_secs()));
        let instant = now.clone();
        let local_peer = peer.clone();
        let sending = running.clone();
        tokio::spawn(async move {
            while sending.load(Ordering::SeqCst) {
                if let Some(event) = receiver.recv().await {
                    if let Err(e) = connection.send(Bytes::from(bincode::serialize(&event).unwrap())).await {
                        error!("send event to node {} error, connection {}, {:?}", local_peer, connection.id(), e);
                        sending.store(false, Ordering::SeqCst);
                    } else {
                        trace!("send event to node {} success, connection {}", local_peer, connection.id());
                        let now = SystemTime::now();
                        let dur = now.duration_since(UNIX_EPOCH).unwrap();
                        instant.store(dur.as_secs(), Ordering::SeqCst);
                    }
                }
            }
            warn!("send message loop exit ------------------------------");
        });

        // send metadata to remote node
        let ping = generator.clone();
        let sender = remote.clone();
        let instant = now.clone();
        let peer_remote = peer.clone();
        let pinging = running.clone();
        tokio::spawn(async move {
            let mut interval = tokio::time::interval(Duration::from_secs(10));
            while pinging.load(Ordering::SeqCst) {
                let now = SystemTime::now();
                let dur = now.duration_since(UNIX_EPOCH).unwrap();
                let early = instant.load(Ordering::SeqCst);
                if dur.as_secs() - early > 10 {
                    let id = ping.lock().await.generate() as u64;
                    if let Err(e) = sender.send(Event::Ping { id }).await {
                        debug!("send auto ping error: {}", e);
                    } else {
                        debug!("auto ping send to {}", peer_remote);
                    }
                }
                let _ = interval.tick().await;
            }
            warn!("ping loop exit ---------------------------------------");
        });

        let timeout = waiting.clone();
        let checking = running.clone();
        tokio::spawn(async move {
            let mut ticker = tokio::time::interval(Duration::from_millis(200));
            while checking.load(Ordering::SeqCst) {
                ticker.tick().await;
                let mut remove = vec![];
                for (key, (time, _sender)) in timeout.read().await.iter() {
                    let key = key.clone();
                    if time.elapsed().unwrap().as_millis() > REMOTE_TIMEOUT {
                        remove.push(key);
                    }
                }
                for key in remove.iter() {
                    if let Some((t, s)) = timeout.write().await.remove(key) {
                        drop(s);
                        warn!("event timed out: key={},time={} ms", key, t.elapsed().unwrap().as_millis())
                    }
                }
            }
            warn!("event waiting timeout handle exit ------------------------------");
        });

        NodeRemote {
            generator,
            waiting,
            running,
            address,
            sender: remote,
            peer,
        }
    }

    pub fn address(&self) -> SocketAddr {
        self.address.clone()
    }

    pub async fn metadata(&self, array: Vec<(PathBuf, Attribute)>) {
        let length = array.len();
        let notify = Event::Metadata(array);
        match self.sender.send(notify).await {
            Ok(()) => info!("send local notify, node length {}", length),
            Err(e) => warn!("send local notify error, {}: {}", length, e),
        };
    }

    pub async fn contacts(&self, array: Vec<SocketAddr>) {
        for addr in array {
            if addr != self.address {
                let message = Event::Contacts(addr.to_string());
                match self.sender.send(message).await {
                    Ok(()) => info!("send local contacts {}", addr),
                    Err(e) => warn!("send local contacts error, {}: {}", addr, e),
                };
            }
        }
    }

    pub async fn open(&self, path: &Path) -> Result<u64, FileSystemError> {
        let key = self.generator.lock().await.generate() as u64;
        self.send(Event::Open {id: key, path: path.to_path_buf()}).await?;
        let (sender, mut receiver) = tokio::sync::mpsc::channel::<Response>(1);
        self.waiting.write().await.insert(key, (SystemTime::now(), sender));
        if let Some(event) = receiver.recv().await {
            match event {
                Response::Open {id, fh, code} => {
                    self.waiting.write().await.remove(&id);
                    if code == 0 {
                        Ok(fh)
                    } else {
                        Err(FileSystemError::RemoteResponseError(code))
                    }
                }
                _ => Err(FileSystemError::RemoteTimeoutError)
            }
        } else {
            Err(FileSystemError::RemoteTimeoutError)
        }
    }

    pub async fn read(&self, fh: u64, offset: u64, size: u32) -> Result<Vec<u8>, FileSystemError> {
        let key = self.generator.lock().await.generate() as u64;
        self.send(Event::Read {id: key, fh, size, offset }).await?;
        let (sender, mut receiver) = tokio::sync::mpsc::channel::<Response>(1);
        self.waiting.write().await.insert(key, (SystemTime::now(), sender));
        if let Some(event) = receiver.recv().await {
            match event {
                Response::Read {id, fh: _fh, code, data} => {
                    self.waiting.write().await.remove(&id);
                    if code == 0 {
                        Ok(data.to_vec())
                    } else {
                        Err(FileSystemError::RemoteResponseError(code))
                    }
                }
                _ => Err(FileSystemError::RemoteTimeoutError)
            }
        } else {
            Err(FileSystemError::RemoteTimeoutError)
        }
    }

    pub async fn write(&self, fh: u64, offset: u64, data: &[u8]) -> Result<u32, FileSystemError> {
        let key = self.generator.lock().await.generate() as u64;
        let bytes = Bytes::copy_from_slice(data);
        self.send(Event::Write {id: key, fh, data: bytes, offset }).await?;
        let (sender, mut receiver) = tokio::sync::mpsc::channel::<Response>(1);
        self.waiting.write().await.insert(key, (SystemTime::now(), sender));
        if let Some(event) = receiver.recv().await {
            match event {
                Response::Write {id, fh: _fh, code, size} => {
                    self.waiting.write().await.remove(&id);
                    if code == 0 {
                        Ok(size)
                    } else {
                        Err(FileSystemError::RemoteResponseError(code))
                    }
                }
                _ => Err(FileSystemError::RemoteTimeoutError)
            }
        } else {
            Err(FileSystemError::RemoteTimeoutError)
        }
    }

    pub async fn release(&self, fh: u64) -> Result<(), FileSystemError> {
        let key = self.generator.lock().await.generate() as u64;
        self.send(Event::Release {id: key, fh }).await?;
        let (sender, mut receiver) = tokio::sync::mpsc::channel::<Response>(1);
        self.waiting.write().await.insert(key, (SystemTime::now(), sender));
        if let Some(event) = receiver.recv().await {
            match event {
                Response::Release {id, fh: _fh, code} => {
                    self.waiting.write().await.remove(&id);
                    if code == 0 {
                        Ok(())
                    } else {
                        Err(FileSystemError::RemoteResponseError(code))
                    }
                }
                _ => Err(FileSystemError::RemoteTimeoutError)
            }
        } else {
            Err(FileSystemError::RemoteTimeoutError)
        }
    }

    pub async fn send(&self, event: Event) -> Result<(), FileSystemError> {
        trace!("send {} event: {:?}", &self.peer, &event);
        if let Err(e) = self.sender.send(event).await {
            error!("send event error: {:?}", e);
            Err(FileSystemError::RemoteConnectError)
        } else {
            Ok(())
        }
    }

    pub fn running(&self) -> bool {
        self.running.load(Ordering::SeqCst)
    }
}

async fn response(id: u64, response: Response, waiters: Arc<RwLock<HashMap<u64, (SystemTime, Sender<Response>)>>>) {
    if let Some((_, sender)) = waiters.write().await.remove(&id) {
        if let Err(e) = sender.send(response).await {
            warn!("send message error: {}", e);
        }
    } else {
        warn!("send message error, no sender found for id {}", id);
    }
}